package process

import java.util

import bean.SensorReading
import org.apache.flink.api.common.functions.{ReduceFunction, RichFlatMapFunction, RichMapFunction}
import org.apache.flink.api.common.state._
import org.apache.flink.api.scala._
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.KeyedProcessFunction
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.util.Collector

/**
  * @Description: TODO QQ1667847363
  * @author: xiao kun tai
  * @date:2021/11/27 12:14
  */
object ProcessTest {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    //file source
    val inputPath: String = "src/main/resources/sensor.txt"
    val fileStream: DataStream[String] = env.readTextFile(inputPath)
    val socketStream = env.socketTextStream("192.168.88.106", 7777)

    //先转换为特定的类型
    val dataStream: DataStream[SensorReading] = socketStream.map(data => {
      val arr = data.split(",")
      SensorReading(arr(0), arr(1).toLong, arr(2).toDouble)
    })

    dataStream
      .keyBy(_.id)
      .process(new MyKeyProcessFunction)

    val warningStream: DataStream[String] = dataStream
      .keyBy(_.id)
      //      .timeWindow(Time.seconds(10))
      .process(new TempIncreaseWarning(10000L))

    warningStream.print("warning")


    env.execute("process function test")
  }

  //实现自定义的KeyProcessFunction
  class TempIncreaseWarning(interval: Long) extends KeyedProcessFunction[String, SensorReading, String] {
    //定义状态：保存上一个温度值进行比较，保存注册定时器的时间戳用于删除
    lazy val lastTempState: ValueState[Double] = getRuntimeContext
      .getState(new ValueStateDescriptor[Double]("temp-state", classOf[Double]))

    lazy val timerTsState: ValueState[Long] = getRuntimeContext
      .getState(new ValueStateDescriptor[Long]("timer-ts", classOf[Long]))


    override def processElement(i: SensorReading, context: KeyedProcessFunction[String, SensorReading, String]#Context, collector: Collector[String]): Unit = {
      //先取出状态
      val lastTemp = lastTempState.value()
      val timerTs = timerTsState.value()

      //判断当前温度值和上次温度值进行比较
      if (i.temperature > lastTemp && timerTs == 0) {
        //如果温度上升，且没有定时器，那么注册当前时间戳10秒之后的定时器
        val ts = context.timerService().currentProcessingTime() + interval
        context.timerService().registerProcessingTimeTimer(ts)
        timerTsState.update(ts)
      } else if (i.temperature < lastTemp) {
        //如果温度下降，那么删除定时器
        context.timerService().deleteProcessingTimeTimer(timerTsState.value())

        timerTsState.clear()
      }
      lastTempState.update(i.temperature)
    }

    override def onTimer(timestamp: Long, ctx: KeyedProcessFunction[String, SensorReading, String]#OnTimerContext, out: Collector[String]): Unit = {
      out.collect("传感器"+ctx.getCurrentKey+"的温度连续"+interval/1000+"秒上升")
      timerTsState.clear()
    }
  }

  //KeyedProcessFunction功能测试
  class MyKeyProcessFunction extends KeyedProcessFunction[String, SensorReading, String] {

    var myState: ValueState[Int] = _

    override def open(parameters: Configuration): Unit = {
      myState = getRuntimeContext
        .getState(new ValueStateDescriptor[Int]("mystate", classOf[Int]))
    }

    override def processElement(i: SensorReading, context: KeyedProcessFunction[String, SensorReading, String]#Context, collector: Collector[String]): Unit = {
      context.getCurrentKey
      context.timestamp()
      context.timerService().currentWatermark()
      context.timerService().currentProcessingTime()
      context.timerService().registerEventTimeTimer(context.timestamp() + 60000L)
      context.timerService().deleteEventTimeTimer(context.timestamp() + 60000L)
    }

    override def onTimer(timestamp: Long, ctx: KeyedProcessFunction[String, SensorReading, String]#OnTimerContext, out: Collector[String]): Unit = {

    }
  }

}
